package com.qkjia.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * @Auther 赵乾坤
 * @Date 2021-10-01 21:27
 * 自定义时间拦截器
 */
public class TimeInterceptor implements ProducerInterceptor<String, String> {

    /**
     * 配置
     * @param configs
     */
    @Override
    public void configure(Map<String, ?> configs) {

    }

    /**
     * 发送时
     * @param record
     * @return
     */
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // 1.取出数据
        String value = record.value();

        // 2.创建一个新的ProducerRecord 对象，，并返回  把时间戳写入消息体的最前部
        return new ProducerRecord<String,String>(record.topic(), record.partition(),
                 record.key(),
                System.currentTimeMillis() + "," + record.value().toString());
    }

    /**
     *
     * @param metadata
     * @param exception
     */
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {

    }

    /**
     * 关闭
     */
    @Override
    public void close() {

    }
}